package com.atguigu.ting.realtime.app.dwd.db;

import com.atguigu.ting.realtime.app.BaseSqlApp;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/11 14:06
 */
public class DwdUserRegister extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdUserRegister().init(
            30041,
            2,
            "DwdUserRegister"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv, "DwdUserRegister");
        // 2. 过滤注册用户记录
        Table userRegister = tEnv.sqlQuery(
            "select " +
                "`data`['id'] id, " +
                "`data`['phone'] phone, " +
                "`data`['is_vip'] is_vip, " +
                "`data`['gender'] gender, " +
                "`data`['create_time'] register_time, " +
                "ts " +
                "from ods_db " +
                "where `database`='ting' " +
                "and `table`='user_info' " +
                "and `type`='insert' ");
        // 3. 写出到 kafka 中
        tEnv.executeSql(
            "create table dwd_user_register(" +
                "id string, " +
                "phone string, " +
                "is_vip string, " +
                "gender string, " +
                "register_time string, " +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_USER_REGISTER));
        userRegister.executeInsert("dwd_user_register");
    }
}
